# uncompyle6 version 3.2.3
# Python bytecode 3.6 (3379)
# Decompiled from: Python 3.6.8 |Anaconda custom (64-bit)| (default, Feb 21 2019, 18:30:04) [MSC v.1916 64 bit (AMD64)]
# Embedded file name: server\routers\api\validator\__init__.py
__author__ = "sanyi"
import re, io, logging, base64, binascii
from collections import defaultdict
from uuid import UUID
import iso8601, dateutil.rrule as dateutil_rrule, datetime, dateutil, dateutil.parser, validate_email, string, json
from api import undefined
from dateutil.tz import tzlocal
from helpers.address import is_url, is_host, is_ip
from helpers.setup_types import is_setup_type_on_premise

__regex_cache = {}
__format_validators = defaultdict(dict)
__type_validators = dict()
HEX_CHARS = string.hexdigits
ALPHANUMERIC_CHARS = string.ascii_letters + string.digits
BASE64_CHARS = ALPHANUMERIC_CHARS + "+/="
_re_time_fm = re.compile("^[0-9]{1,2}:[0-9]{1,2}(:[0-9]{1,2})?$")
_re_license = re.compile(
    "^[0-9A-Z]{4}-[0-9A-Z]{4}-[0-9A-Z]{4}-[0-9A-Z]{4}$", re.IGNORECASE
)
logger = logging.getLogger("server.routers.api.validator")


class ValidationError:
    def __init__(self, code, param_path):
        self.code = code
        self.param_path = param_path

    def to_dict(self):
        return self.__dict__

    def __repr__(self):
        return repr(self.to_dict())

    def __str__(self):
        return self.__repr__()


class ValueNotFound(ValidationError):
    def __init__(self, param_path):
        super().__init__("not_found", param_path)


class ExpectedTypeError(ValidationError):
    def __init__(self, expected_type, param_path):
        super().__init__("invalid_type", param_path)
        self.expected_type = expected_type


class InvalidCondition(ValidationError):
    def __init__(self, condition, param_path):
        super().__init__("invalid_condition", param_path)
        self.condition = condition


class ExtraProperties(ValidationError):
    def __init__(self, name, param_path):
        super().__init__("extra_properties", param_path)
        self.extra_property = re.sub("[^a-zA-Z0-9]", "_", name)


class FormatError(ValidationError):
    def __init__(self, expected_format, param_path):
        super().__init__("format", param_path)
        self.expected_format = expected_format


def validate_format_uuid4(value, param_path):
    try:
        UUID(value, version=4)
        return (None, value)
    except ValueError:
        return (FormatError("uuid4", param_path), None)


def validate_format_empty(value, param_path):
    if value == "":
        return (None, value)
    else:
        return (FormatError("empty", param_path), None)


def validate_format_null(value, param_path):
    if value is None:
        return (None, value)
    else:
        return (FormatError("null", param_path), None)


def validate_format_email(value, param_path):
    try:
        if 5 > len(value) > 254:
            return (FormatError("email", param_path), None)
        is_valid = validate_email.validate_email(value)
        if not is_valid:
            return (FormatError("email", param_path), None)
    except Exception as e:
        print(e)
        return (FormatError("email", param_path), None)

    return (None, value)


def validate_format_sha256(value, param_path):
    if len(value) != 64:
        return (FormatError("sha256", param_path), None)
    elif not all((c in HEX_CHARS for c in value)):
        return (FormatError("sha256", param_path), None)
    else:
        return (None, value)


def validate_format_hex(value, param_path):
    if not all((c in HEX_CHARS for c in value)):
        return (FormatError("hex", param_path), None)
    else:
        return (None, value)


def validate_format_md5(value, param_path):
    if len(value) != 32:
        return (FormatError("md5", param_path), None)
    elif not all((c in HEX_CHARS for c in value)):
        return (FormatError("md5", param_path), None)
    else:
        return (None, value)


def validate_format_rrule(value, param_path):
    if len(value) > 255:
        return (FormatError("rrule", param_path), None)
        try:
            rule = dateutil_rrule.rrulestr(value)
        except ValueError:
            return (FormatError("rrule", param_path), None)

        if rule._freq not in (
            dateutil_rrule.DAILY,
            dateutil_rrule.WEEKLY,
            dateutil_rrule.MONTHLY,
            dateutil_rrule.YEARLY,
        ):
            return (InvalidCondition("invalid rrule frequency", param_path), None)
        rrule_set = dateutil_rrule.rruleset()
        rrule_set.rrule(rule)
        try:
            rrule_set.after(
                datetime.datetime.now(tzlocal()) - (datetime.timedelta(days=1)), True
            )
        except TypeError:
            return (InvalidCondition("invalid rrule dates", param_path), None)
        except Exception as e:
            print(e)
            return (InvalidCondition("invalid rrule format", param_path), None)

        return (None, value)


def validate_format_datetime_iso_8601(value, param_path):
    """
    does type conversion to datetime
    :param value:
    :param param_path:
    :return:
    """
    try:
        return (None, iso8601.parse_date(value))
    except iso8601.iso8601.ParseError:
        return (FormatError("date-time", param_path), None)


def validate_format_int32(value, param_path):
    if value > 2147483647 or value < -2147483648:
        return (FormatError("int32", param_path), None)
    else:
        return (None, value)


def validate_format_int_unsigned(value, param_path):
    if value >= 0:
        return (FormatError("unsigned", param_path), None)
    else:
        return (None, value)


def validate_format_url(value, param_path):
    if len(value) > 2048:
        return (InvalidCondition("max_length", param_path), None)
    elif not is_url(value):
        return (FormatError("url", param_path), None)
    else:
        return (None, value)


def validate_format_host(value, param_path):
    if len(value) > 255:
        return (InvalidCondition("max_length", param_path), None)
    elif not is_host(value):
        return (FormatError("host", param_path), None)
    else:
        return (None, value)


def validate_format_ip(value, param_path):
    if len(value) > 15:
        return (InvalidCondition("max_length", param_path), None)
    elif not is_ip(value):
        return (FormatError("ip", param_path), None)
    else:
        return (None, value)


def validate_format_tag(value, param_path):
    if len(value) > 64:
        return (InvalidCondition("max_length", param_path), None)
    else:
        return (None, value)


def validate_format_search(value, param_path):
    if len(value) > 1024:
        return (InvalidCondition("max_length", param_path), None)
    else:
        return (None, value)


def validate_format_time(value, param_path):
    if len(value) > 8:
        return (InvalidCondition("max_length", param_path), None)
    else:
        if _re_time_fm.fullmatch(value) is None:
            return (FormatError("time", param_path), None)
            try:
                dateutil.parser.parse(value)
            except ValueError:
                return (FormatError("time", param_path), None)

        return (None, value)


def validate_format_license_key(value, param_path):
    if _re_license.fullmatch(value) is None:
        return (FormatError("license_key", param_path), None)
    else:
        return (None, value.upper())


def validate_format_any(value, param_path):
    return (None, value)


def validate_format_json(value, param_path):
    try:
        value = json.loads(value)
    except Exception:
        return (FormatError("json", param_path), None)

    return (None, value)


def validate_format_base32_trimmed(value, param_path):
    for i in range(0, 7):
        try:
            base64.b32decode(value, True)
            break
        except binascii.Error:
            value += "="

    else:
        return (FormatError("invalid base32", param_path), None)

    return (None, value)


def validate_format_otp(value, param_path):
    if len(value) != 6:
        return (InvalidCondition("invalid otp size", param_path), None)
    elif not value.isdigit():
        return (FormatError("invalid otp", param_path), None)
    else:
        return (None, value)


__format_validators["string"]["uuid4"] = validate_format_uuid4
__format_validators["string"]["uuid"] = validate_format_uuid4
__format_validators["string"]["empty"] = validate_format_empty
__format_validators["string"]["null"] = validate_format_null
__format_validators["string"]["email"] = validate_format_email
__format_validators["string"]["sha256"] = validate_format_sha256
__format_validators["string"]["date-time"] = validate_format_datetime_iso_8601
__format_validators["string"]["dateTime"] = validate_format_datetime_iso_8601
__format_validators["string"]["hex"] = validate_format_hex
__format_validators["string"]["md5"] = validate_format_md5
__format_validators["string"]["rrule"] = validate_format_rrule
__format_validators["string"]["any"] = validate_format_any
__format_validators["integer"]["int32"] = validate_format_int32
__format_validators["long"]["int32"] = validate_format_int32
__format_validators["long"]["unsigned"] = validate_format_int_unsigned
__format_validators["integer"]["unsigned"] = validate_format_int_unsigned
__format_validators["string"]["url"] = validate_format_url
__format_validators["string"]["host"] = validate_format_host
__format_validators["string"]["ip"] = validate_format_ip
__format_validators["string"]["tag"] = validate_format_tag
__format_validators["string"]["search"] = validate_format_search
__format_validators["string"]["time"] = validate_format_time
__format_validators["string"]["license_key"] = validate_format_license_key
__format_validators["string"]["json"] = validate_format_json
__format_validators["string"]["base32_trimmed"] = validate_format_base32_trimmed
__format_validators["string"]["otp"] = validate_format_otp


def validate_format_header(value, param_path):
    max_length = 1024
    if is_setup_type_on_premise():
        max_length = 10240
    if len(value) > max_length:
        return (InvalidCondition("max_length", param_path), None)
    else:
        return (None, value)


__format_validators["string"]["header"] = validate_format_header


def validate_format_dummy(value, param_path):
    logger.warning("validation not implemented yet: %s", param_path)
    if len(value) > 512:
        return (InvalidCondition("max_length", param_path), None)
    else:
        return (None, value)


__format_validators["string"]["filename"] = validate_format_dummy
__format_validators["string"]["path_match"] = validate_format_dummy


def register_format_validator(type_name, format_name, validator):
    __format_validators[type_name][format_name] = validator


def validate_format(value, type_name, format_name, param_path):
    if format_name is None:
        return (None, value)
    else:
        e, v = (None, None)
        for fn in format_name.split("|"):
            e, v = __format_validators[type_name][fn](value, param_path)
            if e is None:
                break

        return (e, v)


def validate_float(value, descriptor, param_path, **_):
    """
    :param value:
    :param descriptor:
    :param param_path:
    :param _:
    :return:
    """
    if isinstance(value, str):
        try:
            value = float(value)
        except ValueError:
            return (ExpectedTypeError("float", param_path), None)

    else:
        if not isinstance(value, float):
            return (ExpectedTypeError("float", param_path), None)
        enum = descriptor.get("enum", None)
        if enum:
            if value not in enum:
                return (InvalidCondition("enum", param_path), None)
            return (None, value)
        min_value = descriptor.get("minimum", None)
        max_value = descriptor.get("maximum", None)
        exclusive_maximum = descriptor.get("exclusiveMaximum", False)
        exclusive_minimum = descriptor.get("exclusiveMinimum", False)
        if min_value is not None:
            if (
                exclusive_minimum is True
                and value < min_value
                or exclusive_maximum is False
                and value <= min_value
            ):
                return (InvalidCondition("minimum", param_path), None)
            if max_value is not None:
                if (
                    exclusive_maximum is True
                    and value > max_value
                    or exclusive_minimum is False
                    and value >= max_value
                ):
                    return (InvalidCondition("maximum", param_path), None)
                error, value = validate_format(
                    value, "float", descriptor.get("format", None), param_path
                )
                if error:
                    return (error, None)
                return (None, value)


__type_validators["number"] = validate_float


def validate_integer(value, descriptor, param_path, **_):
    """
    :param value:
    :param descriptor:
    :param param_path:
    :param _:
    :return:
    """
    if isinstance(value, str):
        try:
            value = int(value)
        except ValueError:
            if not value.lower().startswith("0x"):
                return (ExpectedTypeError("integer", param_path), None)
                try:
                    value = int(value, 16)
                except ValueError:
                    return (ExpectedTypeError("integer", param_path), None)

    else:
        if not isinstance(value, int):
            return (ExpectedTypeError("integer", param_path), None)
        enum = descriptor.get("enum", None)
        if enum:
            if value not in enum:
                return (InvalidCondition("enum", param_path), None)
            return (None, value)
        min_value = descriptor.get("minimum", None)
        max_value = descriptor.get("maximum", None)
        exclusive_maximum = descriptor.get("exclusiveMaximum", False)
        exclusive_minimum = descriptor.get("exclusiveMinimum", False)
        if max_value is not None:
            if (
                exclusive_maximum is True
                and value >= max_value
                or exclusive_maximum is False
                and value > max_value
            ):
                return (InvalidCondition("maximum", param_path), None)
            if min_value is not None:
                if (
                    exclusive_minimum is True
                    and value <= min_value
                    or exclusive_minimum is False
                    and value < min_value
                ):
                    return (InvalidCondition("minimum", param_path), None)
                multiple_of = descriptor.get("multipleOf", None)
                if multiple_of is not None:
                    if value % multiple_of:
                        return (ValidationError("multipleOf", param_path), None)
                    error, value = validate_format(
                        value, "integer", descriptor.get("format", None), param_path
                    )
                    if error:
                        return (error, None)
                    return (None, value)


__type_validators["integer"] = validate_integer
__type_validators["long"] = validate_integer


def validate_string(value, descriptor, path_param, **_):
    if value is None:
        value_format = descriptor.get("format")
        if value_format:
            if "null" in value_format.split("|"):
                return (None, value)
            if "any" in value_format.split("|"):
                return (None, value)
            return (ExpectedTypeError("string", path_param), None)
        if not isinstance(value, str):
            return (ExpectedTypeError("string", path_param), None)
        enum = descriptor.get("enum", None)
        if enum:
            if value not in enum:
                return (InvalidCondition("enum", path_param), None)
            return (None, value)
        str_len = len(value)
        fixed_length = descriptor.get("length", None)
        if fixed_length is not None:
            if str_len != fixed_length:
                return (InvalidCondition("length", path_param), None)
            min_length = descriptor.get("minLength", 0)
            if str_len < min_length:
                return (InvalidCondition("minLength", path_param), None)
            max_length = descriptor.get("maxLength", None)
            if max_length is not None:
                if str_len > max_length:
                    return (InvalidCondition("maxLength", path_param), None)
                pattern = descriptor.get("pattern", None)
                if pattern:
                    re_pattern = __regex_cache.get(pattern, None)
                    if re_pattern is None:
                        re_pattern = re.compile(pattern)
                    if re_pattern.fullmatch(value) is None:
                        return (InvalidCondition("pattern", path_param), None)
                    error, value = validate_format(
                        value, "string", descriptor.get("format", None), path_param
                    )
                    if error:
                        return (error, None)
                    return (None, value)


__type_validators["string"] = validate_string


def validate_boolean(value, descriptor, path_param, **_):
    """
    If value is string or int attempts type conversion.
    accepted string values are 'true', '1', 'yes', 'false', '0', 'no' case insensitive
    accepted int values are 0 for False, any!=0 for True
    :param value:
    :param descriptor:
    :param path_param:
    :param _:
    :return:
    """
    if value is None:
        value_format = descriptor.get("format")
        if value_format:
            if "null" in value_format.split("|"):
                return (None, value)
            return (ExpectedTypeError("boolean", path_param), None)
        if isinstance(value, str):
            value = value.lower() in frozenset({"1", "yes", "true"}) and True
        else:
            if value.lower() in frozenset({"false", "no", "0"}):
                value = False
            else:
                if isinstance(value, int):
                    value = value != 0
            if not isinstance(value, bool):
                return (ExpectedTypeError("boolean", path_param), None)
            enum = descriptor.get("enum", None)
            if enum:
                if value not in enum:
                    return (InvalidCondition("enum", path_param), None)
                return (None, value)


__type_validators["boolean"] = validate_boolean


def validate_array(value, descriptor, path_param, **kwargs):
    if value is None:
        value_format = descriptor.get("format")
        if value_format:
            if "null" in value_format.split("|"):
                return (None, value)
            return (ExpectedTypeError("array", path_param), None)
        if not isinstance(value, (list, tuple)):
            return (ExpectedTypeError("array", path_param), None)
        max_items = descriptor.get("maxItems", None)
        if max_items is not None:
            if max_items < len(value):
                return (InvalidCondition("maxItems", path_param), None)
        min_items = descriptor.get("minItems", None)
        if min_items is not None:
            if min_items > len(value):
                return (InvalidCondition("minItems", path_param), None)
            unique_items = descriptor.get("uniqueItems", False)
            if unique_items:
                for i in range(0, len(value) - 1):
                    for j in range(i + 1, len(value)):
                        if value[i] == value[j]:
                            return (InvalidCondition("uniqueItems", path_param), None)

            items = descriptor.get("items", None)
            additional_items = descriptor.get("additionalItems", False)
            if additional_items is True:
                logger.warning(
                    "no array item validation [additional_items=True] for: %s",
                    path_param,
                )
            else:
                if items is None:
                    logger.warning("no array item validation for: %s", path_param)
        if isinstance(items, (list, tuple)):
            if additional_items is False:
                if len(value) != len(items):
                    return (InvalidCondition("additionalItems", path_param), None)
                len_items = len(items)
                for index in range(0, len(value)):
                    if index >= len_items:
                        if isinstance(additional_items, dict):
                            error, param_value = validate_value(
                                path_param="%s[%d]" % (path_param, index), **kwargs
                            )
                            if error:
                                return (error, None)
                            value[index] = param_value
                        else:
                            if additional_items is False:
                                return (
                                    InvalidCondition(
                                        "additionalItems",
                                        "%s[%d]" % (path_param, index),
                                    ),
                                    None,
                                )
                    else:
                        error, param_value = validate_value(
                            path_param="%s[%d]" % (path_param, index), **kwargs
                        )
                        if error:
                            if isinstance(additional_items, dict):
                                error, param_value = validate_value(
                                    path_param="%s[%d]" % (path_param, index), **kwargs
                                )
                                if error:
                                    return (error, None)
                            else:
                                return (error, None)
                            value[index] = param_value

            else:
                if isinstance(items, dict):
                    for index in range(0, len(value)):
                        error, param_value = validate_value(
                            path_param="%s[%d]" % (path_param, index), **kwargs
                        )
                        if error:
                            if isinstance(additional_items, dict):
                                error, param_value = validate_value(
                                    path_param="%s[%d]" % (path_param, index), **kwargs
                                )
                                if error:
                                    return (error, None)
                            else:
                                return (error, None)
                            value[index] = param_value

                else:
                    if not False:
                        raise AssertionError
                return (None, value)


__type_validators["array"] = validate_array


def validate_object(value, descriptor, path_param, **kwargs):
    if value is None:
        if kwargs.get("accept_null_objects"):
            return (None, value)
        if value is None:
            value_format = descriptor.get("format")
            if value_format:
                if "null" in value_format.split("|"):
                    return (None, value)
                return (ExpectedTypeError("boolean", path_param), None)
            if not isinstance(value, dict):
                return (ExpectedTypeError("object", path_param), None)
            max_properties = descriptor.get("maxProperties", None)
            if max_properties is not None:
                if len(value) > max_properties:
                    return (InvalidCondition("maxProperties", path_param), None)
                min_properties = descriptor.get("minProperties", None)
                if min_properties is not None:
                    if len(value) < min_properties:
                        return (InvalidCondition("minProperties", path_param), None)
                    problems = []
                    for param_name, param_description in descriptor[
                        "properties"
                    ].items():
                        param_value = value.get(param_name, undefined)
                        error, param_value = validate_value(
                            path_param=path_param + "." + param_name, **kwargs
                        )
                        if error:
                            if isinstance(error, (list, tuple)):
                                for e in error:
                                    problems.append(e)

                            else:
                                problems.append(error)
                        elif param_value != undefined:
                            value[param_name] = param_value

                    for param_name in list(value.keys()):
                        if param_name not in descriptor["properties"]:
                            problems.append(ExtraProperties(param_name, path_param))
                            del value[param_name]

                    return (problems, value)


__type_validators["object"] = validate_object


def validate_file_upload(value, descriptor, path_param, **kwargs):
    if value is None:
        if kwargs.get("accept_null_objects"):
            return (None, value)
        if not isinstance(value, io.IOBase):
            return (ExpectedTypeError("object", path_param), None)
        return (None, value)


__type_validators["file"] = validate_file_upload


def validate_value(value, descriptor, path_param="", **kwargs):
    """
    During validation problems are collected, every value/sub-value with problems is reset to None
    value may suffer modifications, ex, because:
         - has no value set (object class) but descriptor had default set
         - was string representation of float, number, boolean and had it's type adjusted
    :param value: the value should be validated
    :param descriptor: the swagger description object of the value should be validated
    :param path_param: optional value for building a composite parameter name in errors
    :param kwargs: for example add_defaults=False will disable defaults population
    :return: (problems, value)
    """
    add_defaults = kwargs.get("add_defaults", True)
    ignore_required = kwargs.get("ignore_required", False)
    if value == undefined:
        if add_defaults:
            value = descriptor.get("default", undefined)
        if value == undefined:
            if not ignore_required:
                if descriptor.get("required", False):
                    return (ValueNotFound(path_param), None)
                return (None, value)
            problems, value = __type_validators[descriptor["type"]](
                value, descriptor, path_param, **kwargs
            )
            return (problems, value)


def register_type_validator(name, validator):
    __type_validators[name] = validator
